package org.lql.chaos.kafka.example.producer;

import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.lql.chaos.kafka.example.callback.CustomProducerCallback;
import org.lql.chaos.kafka.example.pojo.BaseMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.nio.charset.StandardCharsets;
import java.util.UUID;

/**
 * @author: liangqinglong
 * @date: 2025-08-10 01:18
 * @description:
 **/
@Service
public class KafkaProducerExampleService {

	@Autowired
	@Qualifier("kafkaTemplateString")
	private KafkaTemplate<String, String> kafkaTemplateString;

	@Autowired
	@Qualifier("kafkaTemplateBytes")
	private KafkaTemplate<String, byte[]> kafkaTemplateBytes;

	@Autowired
	@Qualifier("kafkaTemplateJson")
	private KafkaTemplate<String, Object> kafkaTemplateJson;

	@Autowired
	@Qualifier("kafkaTemplateKryo")
	private KafkaTemplate<String, Object> kafkaTemplateCustom;

	// 统一方法：设置messageId，如果是BaseMessage子类
	private <T> T prepareMessage(T data) {
		if (data instanceof BaseMessage) {
			((BaseMessage) data).setMessageId(UUID.randomUUID().toString());
		}
		return data;
	}

	// 统一发送方法
	private <K, V> void sendMessage(KafkaTemplate<K, V> kafkaTemplate, String topic, K key, V value, String partitionStrategy) {
		ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, key, value);
		producerRecord.headers().add("partition-strategy", partitionStrategy.getBytes(StandardCharsets.UTF_8));
		kafkaTemplate.send(producerRecord).addCallback(new CustomProducerCallback<>(key, value));
	}

	/**
	 * 发送简单消息，使用 String 序列化器
	 *
	 * @param topic   主题
	 * @param message 消息
	 */
	public void sendSimpleMessageByStringSerializer(String topic, String key, String message, String partitionStrategy) {
		sendMessage(kafkaTemplateString, topic, key, message, partitionStrategy);
	}

	/**
	 * 发送复杂对象消息，使用 String 序列化器
	 *
	 * @param topic 主题
	 * @param data  数据
	 */
	public <T> void sendComplexObjectMessageByStringSerializer(String topic, String key, T data, String partitionStrategy) {
		T prepared = prepareMessage(data);
		String message = JSONObject.toJSONString(prepared);
		sendMessage(kafkaTemplateString, topic, key, message, partitionStrategy);
	}

	/**
	 * 发送简单消息，使用 Bytes 序列化器
	 *
	 * @param topic   主题
	 * @param message 消息
	 */
	public void sendSimpleMessageByBytesSerializer(String topic, String key, byte[] message, String partitionStrategy) {
		sendMessage(kafkaTemplateBytes, topic, key, message, partitionStrategy);
	}

	/**
	 * 发送复杂对象消息，使用 Bytes 序列化器
	 *
	 * @param topic 主题
	 * @param data  数据
	 */
	public <T> void sendComplexObjectMessageByBytesSerializer(String topic, String key, T data, String partitionStrategy) {
		T prepared = prepareMessage(data);
		byte[] bytes = JSONObject.toJSONBytes(prepared);
		sendMessage(kafkaTemplateBytes, topic, key, bytes, partitionStrategy);
	}

	/**
	 * 发送简单消息，使用 Json 序列化器
	 *
	 * @param topic   主题
	 * @param message 消息
	 */
	public void sendSimpleMessageByJsonSerializer(String topic, String key, Object message, String partitionStrategy) {
		sendMessage(kafkaTemplateJson, topic, key, message, partitionStrategy);
	}

	/**
	 * 发送复杂对象消息，使用 Json 序列化器
	 *
	 * @param topic 主题
	 * @param data  数据
	 */
	public <T> void sendComplexObjectMessageByJsonSerializer(String topic, String key, T data, String partitionStrategy) {
		T prepared = prepareMessage(data);
		sendMessage(kafkaTemplateJson, topic, key, prepared, partitionStrategy);
	}

	/**
	 * 发送简单消息，使用自定义序列化器
	 *
	 * @param topic   主题
	 * @param message 消息
	 */
	public void sendSimpleMessageByCustomSerializer(String topic, String key, Object message, String partitionStrategy) {
		sendMessage(kafkaTemplateCustom, topic, key, message, partitionStrategy);
	}

	/**
	 * 发送复杂对象消息，使用自定义序列化器
	 *
	 * @param topic 主题
	 * @param data  数据
	 */
	public <T> void sendComplexObjectMessageByCustomSerializer(String topic, String key, T data, String partitionStrategy) {
		T prepared = prepareMessage(data);
		sendMessage(kafkaTemplateCustom, topic, key, prepared, partitionStrategy);
	}

}
